package cgl.narada.benchmark.security;

import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Properties;
import java.util.Random;
import java.util.Vector;

import org.apache.log4j.Logger;

import cgl.narada.event.NBEvent;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.matching.Profile;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.ClientService;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.client.NBEventListener;
import cgl.narada.service.client.SessionService;
import cgl.narada.util.SystemInit;
import cgl.narada.util.SystemNotInitializedException;

/**
 * <br>
 * Created on Apr 20, 2006
 * 
 * @author Harshawardhan Gadgil (hgadgil@grids.ucs.indiana.edu)
 */
public class BenchmarkListener implements NBEventListener {
    static Logger log = Logger.getLogger("BenchmarkListener");

    private static BenchmarkListener instance = new BenchmarkListener();

    private BenchmarkListener() {}

    public static final String BENCHMARK_COMMIT = "benchmark/commit";
    private static final boolean writeThrough = true;
    // 
    private Hashtable ht = new Hashtable();

    // --------------------
    private String configFileLocation = null;
    private EventConsumer consumer;

    public static String tcpPort;

    private Profile profile;

    private ClientService clientService;

    public static BenchmarkListener getInstance() {
        return instance;
    }

    public void init() {
        int entityId = (new Random()).nextInt();
        // Connect to broker and listen to
        // benchmark/commit topic
        Properties prop = new Properties();
        prop.put("hostname", "localhost");

        // IMP: Might conflict w/ other ppl's entity IDs Maintain constant for
        // now

        prop.put("portnum", tcpPort);

        try {
            configFileLocation = SystemInit
                    .getInitData(SystemInit.SYS_SERVICE_CONFIG_PATH);
        } catch (SystemNotInitializedException e1) {
            e1.printStackTrace();
        }

        log.info("Connecting using TCP @ " + prop.getProperty("hostname") + ":"
                + tcpPort + " EntityID[" + entityId + "]");

        try {
            clientService = SessionService.getClientService(entityId,
                    configFileLocation);

            clientService.initializeBrokerCommunications(prop, "tcp");
            profile = clientService.createProfile(
                    TemplateProfileAndSynopsisTypes.STRING, BENCHMARK_COMMIT);
            consumer = clientService.createEventConsumer(this);

            consumer.subscribeTo(profile);
        } catch (ServiceException e) {
            log.error("", e);
        }
    }

    public void register(String key) {
        if (!writeThrough)
            ht.put(key, new Vector());
    }

    public void set(String key, String value) {
        if (writeThrough) {
            FileOutputStream fout = null;
            if (ht.containsKey(key)) {
                fout = (FileOutputStream) ht.get(key);
            } else {
                try {
                    fout = new FileOutputStream("/home/hgadgil/" + key);
                    ht.put(key, fout);
                } catch (FileNotFoundException e) {
                    e.printStackTrace();
                }
            }

            try {
                fout.write((value + "\n").getBytes());
            } catch (IOException e) {
                e.printStackTrace();
            }
        } else
            ((Vector) ht.get(key)).add(value);
    }

    public void onEvent(NBEvent event) {
        // Start committing files

        if (!writeThrough) {
            for (Enumeration e = ht.keys(); e.hasMoreElements();) {
                String key = (String) e.nextElement();

                Vector v = (Vector) ht.get(key);
                if (v.size() > 0) {
                    System.out.println("Committing: " + key + " --> ["
                            + v.size() + "] entries !");

                    // Now write to file

                    try {
                        FileOutputStream fout = new FileOutputStream(
                                "/home/hgadgil/" + key);

                        for (int i = 0; i < v.size(); i++) {
                            fout.write(((String) v.elementAt(i) + "\n")
                                    .getBytes());
                        }
                        fout.flush();
                        fout.close();
                    } catch (IOException x) {
                        x.printStackTrace();
                    }
                }
            }
        }

        System.exit(0);
    }
}
